package mockdata;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class demotest {

    public static void main(String[] args) throws InterruptedException {
        Properties kafkaproperties = new Properties();
        kafkaproperties.setProperty("bootstrap.servers", "master3:9092,master4:9092,master5:9092");
        kafkaproperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        kafkaproperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        kafkaproperties.put("acks", "all");
        kafkaproperties.put("batch.size",1024);  //这里1k,默认是16k。当发送数据的时候，不用等到batchsize16k才发送,这样可以更快的发送，但吞吐有影响
        kafkaproperties.put("linger.ms",1);//当>0时 ，延时会增加，但会减少发送频率，提供吞吐。只有当batch填满或者达到这个时间，就会发送方数据。
        KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(kafkaproperties);


        for (int i = 0; i <10 ; i++) {
            String  message=i+1+"";
            KafkaProducer.send(new ProducerRecord<>("aaaa", message));
            Thread.sleep(10);
        }
    }
}
